package com.shellyan.gmall.rt.app

import java.util

import com.alibaba.fastjson.JSON
import com.atguigu.realtime.gmall.common.Constant
import com.shellyan.gmall.rt.bean.{AlertInfo, EventLog}
import org.apache.spark.streaming.{Minutes, Seconds}
import org.apache.spark.streaming.dstream.DStream

import scala.util.control.Breaks._

/**
 * @author Shelly An
 * @create 2020/9/7 14:17
 *         同一设备，5分钟内三次及以上用不同账号登录并领取优惠劵，并且在登录到领劵过程中没有浏览商品。
 *         同时达到以上要求则产生一条预警日志。 同一设备，每分钟只记录一次预警。
 *
 *
 *         同一设备 ： 按mid分组
 *         5分钟以内，每隔6s更新一次预警：窗口  实现方式 1.窗口算子 2. .window
 *         元素：窗口长度、窗口步长
 *         三次及以上用不同账号登录并领取优惠劵：统计领取优惠券的用户数
 *         没有浏览商品：过滤
 *         每分钟只记录一次预警: 让es来负责去重，es幂等性
 */
object AlertApp extends BaseApp {
  override val master: String = "local[2]"
  override val appName: String = "AlertApp"
  override val batchTime: Int = 3
  override val topics: Set[String] = Set(Constant.EVENT_TOPIC)
  override val groupId: String = "AlertApp"

  //{"logType":"event","area":"guangdong","uid":"6696","eventId":"coupon","itemId":4,"os":"android","nextPageId":47,"appId":"gmall","mid":"mid_453","pageId":7,"ts":1599460000415}
  override def run(sourceStream: DStream[String]): Unit = {
    //带窗口的流
    val eventLogStream: DStream[(String, EventLog)] = sourceStream.map(json => {
      val log: EventLog = JSON.parseObject(json, classOf[EventLog])
      (log.mid, log)
    }).window(Minutes(5), Seconds(6))

    val alertInfoStream: DStream[(Boolean, AlertInfo)] = eventLogStream
      .groupByKey()
      .map {
        /*
        Iterable遍历，计算想要的数据，一般会进行冗余设计，尽量提取全，后续有可能想用
         */
        case (mid, it: Iterable[EventLog]) =>
          //存储在当前设备领取优惠券的用户id  用java的集合，因为要写入es
          val uids = new util.HashSet[String]()
          //在当前设备的所有其他操作时间
          val events = new util.ArrayList[String]()
          //存储优惠券所在的商品
          val items = new util.HashSet[String]()
          //布尔值来表示是否浏览商品
          var isBrowser = false
          breakable {
            it.foreach(eventlog => {
              events.add(eventlog.eventId)
              //模式匹配 替代if 性能更好
              eventlog.eventId match {
                //优惠券
                case "coupon" =>
                  uids.add(eventlog.uid)
                  items.add(eventlog.itemId)
                //浏览商品
                case "clickItem" =>
                  //如果5分钟内有浏览商品，则不做预警处理
                  //return也是抛异常
                  isBrowser = true
                  //后面一定不会预警，可以不用处理了
                  break
                //其他事件
                case _ =>
              }
            })
          }


          (
            uids.size() >= 3 && !isBrowser, //(true,具体预警信息)   预警   (false,具体预警信息)  不预警
            AlertInfo(mid, uids, items, events, System.currentTimeMillis())
          )
      }

    alertInfoStream
      .filter(_._1)
      .map(_._2)
      .foreachRDD(rdd => {
      //把数据写入es，在Kibana中提前创建好index
      //方式1：
      /*      rdd.foreachPartition(infoIt=>{
              ESUtil
                .insertBulk("gmall_coupon_alert",
                  infoIt.map(info => (s"${info.mid}_${info.ts/1000/60}",info)))
            })*/
      //隐式转换函数
      //implicit val rdd2RichES = (rdd: RDD[AlertInfo]) => new RichES(rdd)
      //隐式转换函数 --升级--> 隐式类
      import com.shellyan.gmall.rt.util.ESUtil._
      rdd.saveToES("gmall_coupon_alert")
    })
  }
}

